package com.bobo.netty.utils.threadUtils;

import com.alibaba.ttl.threadpool.TtlExecutors;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.Validate;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.Maps;
import java.util.Map;
import java.util.concurrent.*;

public class ThreadPoolManager {
    private static Logger log = LoggerFactory.getLogger(ThreadPoolManager.class);
    private String threadNamePrefix;
    private int corePoolSize;
    private int maxPoolSize;
    private int queueCapacity = Integer.MAX_VALUE;
    private RejectedExecutionHandler rejectedExecutionHandler = new MyThreadPoolExecutor.AbortPolicy();
    private ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1);
    private ExecutorService threadPoolExecutor;
    private IThreadConfigChange iThreadConfigChange;
    private volatile boolean started = false;


    protected ThreadPoolManager(String threadNamePrefix, int poolSize) {
        this(threadNamePrefix, poolSize, poolSize);
    }

    protected ThreadPoolManager(String threadNamePrefix, int poolSize, int maxSize) {
        this.threadNamePrefix = threadNamePrefix;
        this.corePoolSize = poolSize;
        this.maxPoolSize = maxSize;
    }

    protected ThreadPoolManager(String threadNamePrefix, int poolSize, IThreadConfigChange iThreadConfigChange) {
        this(threadNamePrefix, poolSize, poolSize);
        this.iThreadConfigChange = iThreadConfigChange;
    }

    protected ThreadPoolManager(String threadNamePrefix, int poolSize, IThreadConfigChange iThreadConfigChange
            , int queueCapacity, RejectedExecutionHandler rejectedExecutionHandler) {
        this(threadNamePrefix, poolSize, iThreadConfigChange);
        this.rejectedExecutionHandler = rejectedExecutionHandler;
        this.threadNamePrefix = threadNamePrefix;
        this.queueCapacity = queueCapacity;
    }

    protected ThreadPoolManager init() {
        return init(true);
    }
    protected ThreadPoolManager init(boolean monitor) {
        if (started) {
            throw new RuntimeException("当前线程池已经启动，无需再启");
        }
        checkParam();
        this.threadNamePrefix = StringUtils.isNotBlank(threadNamePrefix) ? threadNamePrefix : "thread-pool-proxy";
        BlockingQueue<Runnable> queue = this.createQueue(this.queueCapacity);
        ThreadFactory threadFactory = new BasicThreadFactory.Builder()
                .namingPattern(this.threadNamePrefix + "-1%s").build();
        long keepAliveSeconds = 60;
        MyThreadPoolExecutor executor = new MyThreadPoolExecutor(
                this.corePoolSize, this.maxPoolSize,
                keepAliveSeconds, TimeUnit.SECONDS,
                queue, threadFactory, rejectedExecutionHandler);
        boolean allowCoreThreadTimeOut = false;
        if (allowCoreThreadTimeOut) {
            executor.allowCoreThreadTimeOut(true);
        }
        this.threadPoolExecutor = TtlExecutors.getTtlExecutorService(executor);
        //通过MCC调整线程池并进行监控
        executorService.scheduleAtFixedRate(() -> {
            if (iThreadConfigChange != null) {
                int coreSize = iThreadConfigChange.getCorePoolSize();
                int maxSize = iThreadConfigChange.getMaxPoolSize();
                if (coreSize != 0 && coreSize != this.getExecutorCorePoolSize()) {
                    log.info("{} change core-size-->{}", this.getThreadNamePrefix(), coreSize);
                    this.setExecutorCorePoolSize(coreSize);
                }
                if (maxSize != 0 && maxSize != this.getExecutorMaxPoolSize()) {
                    log.info("{} change max-size-->{}", this.getThreadNamePrefix(), maxSize);
                    this.setExecutorMaxPoolSize(maxSize);
                }
            }
            //cat-client event 当最大线程数满并且队列满时打点,可设置报警
            if (getLocalOriginThreadPool().getActiveCount() == getLocalOriginThreadPool().getMaximumPoolSize()
                    && getLocalOriginThreadPool().getQueue().remainingCapacity() == 0) {
                String join = StringUtils.join(
                        "activeCount:", getLocalOriginThreadPool().getActiveCount(),
                        " queueSize:", getLocalOriginThreadPool().getQueue().size());
                log.error("ThreadPool-Full:{},{}", this.getThreadNamePrefix(), join);
            }
        }, 0, 60, TimeUnit.SECONDS);
        //固定打印部分线程池信息或者上报
//        executorService.scheduleAtFixedRate(this::monitor, 60, 60, TimeUnit.SECONDS);

//        if (monitor) {
//            //Cat Heartbeat
//            StatusExtensionRegister.getInstance().register(new StatusExtension() {
//                @Override
//                public String getDescription() {
//                    return getThreadNamePrefix();
//                }
//
//                @Override
//                public String getId() {
//                    return getThreadNamePrefix();
//                }
//
//                @Override
//                public Map<String, String> getProperties() {
//                    return monitor();
//                }
//            });
//        }

        started = true;
        //关闭前处理
        Runtime.getRuntime().addShutdownHook(new Thread(this::monitor));
        return this;

    }

    private Map<String, String> monitor() {
        try {
            Map<String, String> map = Maps.newHashMap();
            map.put("core-size", String.valueOf(getLocalOriginThreadPool().getCorePoolSize()));
            map.put("max-size", String.valueOf(getLocalOriginThreadPool().getMaximumPoolSize()));
            map.put("active-size", String.valueOf(getLocalOriginThreadPool().getActiveCount()));
            map.put("queue-size", String.valueOf(getLocalOriginThreadPool().getQueue().size()));
            log.info("thread-monitor:{},{}", getThreadNamePrefix(), map);
            return map;
        } catch (Exception e) {
            log.error("线程池处理异常", e);
            return null;
        }
    }

    private void checkParam() {
        Validate.notEmpty(threadNamePrefix, "threadNamePrefix is not allow empty");
    }

    private BlockingQueue<Runnable> createQueue(int queueCapacity) {
        return (BlockingQueue) (queueCapacity > 0 ? new LinkedBlockingQueue(queueCapacity) : new SynchronousQueue());
    }

    private ExecutorService getThreadPoolExecutor() throws IllegalStateException {
        Validate.isTrue(this.threadPoolExecutor != null, "ThreadPoolTaskExecutor not initialized");
        return this.threadPoolExecutor;
    }

    private void setExecutorCorePoolSize(int corePoolSize) {
        getLocalOriginThreadPool().setCorePoolSize(corePoolSize);
    }

    private ThreadPoolExecutor getLocalOriginThreadPool() {
        return (MyThreadPoolExecutor) TtlExecutors.unwrap(this.threadPoolExecutor);
    }

    private int getExecutorCorePoolSize() {
        return this.corePoolSize;
    }

    private void setExecutorMaxPoolSize(int maxPoolSize) {
        getLocalOriginThreadPool().setMaximumPoolSize(maxPoolSize);
    }

    private int getExecutorMaxPoolSize() {
        return this.maxPoolSize;
    }

    private String getThreadNamePrefix() {
        return threadNamePrefix;
    }


    public Future<?> submit(Runnable task) {
        return getThreadPoolExecutor().submit(task);
    }

    public <T> Future<T> submit(Callable<T> task) {
        return getThreadPoolExecutor().submit(task);
    }

    static class MyThreadPoolExecutor extends ThreadPoolExecutor {

        public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }

        public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        }

        public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        }

        public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        }

        @Override
        protected void afterExecute(Runnable r, Throwable t) {
            if (t != null) {
                log.error("任务执行异常", t);
            }
            super.afterExecute(r, t);
        }
    }
}
